package com.huidu.bitrade.consumer;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson2.JSON;
import com.huidu.bitrade.component.CoinExchangeRate;
import com.huidu.bitrade.constant.NettyCommand;
import com.huidu.bitrade.entity.*;
import com.huidu.bitrade.handler.NettyHandler;
import com.huidu.bitrade.job.ExchangePushJobNew;
import com.huidu.bitrade.processor.CoinProcessor;
import com.huidu.bitrade.processor.CoinProcessorFactory;
import com.huidu.bitrade.service.ExchangeCoinService;
import com.huidu.bitrade.service.ExchangeOrderService;
import com.huidu.bitrade.service.ExchangeTradeHandler;
import com.huidu.bitrade.util.MessageResult;
import com.huidu.bitrade.util.NettyPushDataUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.support.TransactionTemplate;

import javax.annotation.PostConstruct;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j(topic = "consumer")
public class ExchangeTradeConsumer {
    @Autowired
    private CoinProcessorFactory coinProcessorFactory;
    @Autowired
    private SimpMessagingTemplate messagingTemplate;
    @Autowired
    private ExchangeOrderService exchangeOrderService;
    @Autowired
    private NettyHandler nettyHandler;
    @Value("${second.referrer.award}")
    private boolean secondReferrerAward;
    @Autowired
    private ExchangeCoinService coinService;
    @Autowired
    private CoinExchangeRate coinExchangeRate;
    @Autowired
    private ExchangePushJobNew pushJob;
    @Autowired
    private TransactionTemplate txTemplate;
    @Autowired
    private NettyPushDataUtil nettyPushDataUtil;

    private ExecutorService executor = new ThreadPoolExecutor(30, 100, 0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>(1024), new ThreadPoolExecutor.AbortPolicy());

    static final RequestConfig HTTP_CONFIG = RequestConfig.custom()
            .setConnectTimeout(200)
            .setConnectionRequestTimeout(200)
            .setSocketTimeout(200).build();

    private static final PoolingHttpClientConnectionManager POOL = new PoolingHttpClientConnectionManager();

    static {
        POOL.setMaxTotal(128);
    }

    private static final CloseableHttpClient HTTP_CLIENT = HttpClients.custom()
            .setDefaultRequestConfig(HTTP_CONFIG)
            .setConnectionManager(POOL)
            .build();


    @PostConstruct
    public void init() {
        txTemplate.setIsolationLevel(TransactionDefinition.ISOLATION_REPEATABLE_READ);
    }

    @KafkaListener(topics = "exchange-trade", containerFactory = "kafkaListenerContainerFactory")
    public void handleTrade(List<ConsumerRecord<String, String>> records) {
        for (int i = 0; i < records.size(); i++) {
            ConsumerRecord<String, String> record = records.get(i);
            executor.submit(new HandleTradeThread(record));
        }
    }

    private void handleComplete(ExchangeOrder order) {
        String symbol = order.getSymbol();
        try {
            MessageResult res = exchangeOrderService.tradeCompleted(order.getOrderId(), order.getTradedAmount(),
                    order.getTurnover());
            log.info("<{} {} {} {}", order.getOrderId(), order.getSymbol(), order.getMemberId(), res.getMessage());

            log.info("handleComplete 订单完成开始推送");
            messagingTemplate.convertAndSend(
                    "/topic/market/order-completed/" + symbol + "/" + order.getMemberId(), order);
            log.info(DateUtil.format(new Date(), DatePattern.NORM_DATETIME_MS_PATTERN));
            //todo 就是告诉前端，发个通知而已，客户端没有处理数据，客户端只是刷新了列表
            nettyHandler.handleOrder(NettyCommand.PUSH_EXCHANGE_ORDER_COMPLETED, order);
            log.info("handleComplete 订单完成开始结束");
        } catch (Exception e) {
            log.error("", e);
        }
    }

    private void handleCancel(ExchangeOrder order) {
        String symbol = order.getSymbol();
        // 调用服务处理
        MessageResult result = exchangeOrderService.cancelOrder(order.getOrderId(), order.getTradedAmount(),
                order.getTurnover());
        log.info("\\ {} {} {} {}", order.getOrderId(), order.getSymbol(), order.getMemberId(), result.getMessage());
        // 推送实时成交
        messagingTemplate.convertAndSend("/topic/market/order-canceled/" + symbol + "/" + order.getMemberId(),
                order);

        //todo 就是告诉前端，发个通知而已，客户端没有处理数据，客户端只是刷新了列表
        nettyHandler.handleOrder(NettyCommand.PUSH_EXCHANGE_ORDER_CANCELED, order);
    }

    private List<CoinThumb> getCoinThumbs() {
        List<ExchangeCoin> coins = coinService.findAllVisible();
        List<CoinThumb> thumbs = new ArrayList<>();
        for (ExchangeCoin coin : coins) {
            CoinProcessor processor = coinProcessorFactory.getProcessor(coin.getSymbol());
            CoinThumb thumb = processor.getThumb();
            if (coin.getBaseSymbol().equals("CNHD")) {
                BigDecimal baseUsdRate = coinExchangeRate.getUsdRate(coin.getBaseSymbol());
                thumb.setTurnover(thumb.getTurnover().multiply(baseUsdRate).setScale(4, RoundingMode.UP));
            }
            thumb.setZone(coin.getZone());
            thumbs.add(thumb);
        }
        return thumbs;
    }

    private void handleTrade(ExchangeTrade trade) {
        String symbol = trade.getSymbol();
        CoinProcessor coinProcessor = coinProcessorFactory.getProcessor(symbol);
        List<ExchangeTrade> successTrades = Collections.emptyList();
        CoinThumb[] thumbs = getCoinThumbs().toArray(new CoinThumb[0]);
        MessageResult<Map.Entry<ExchangeTradeHandler, ExchangeTradeHandler>> result = txTemplate.execute((stat) -> {
            try {
                return exchangeOrderService.processExchangeTrade(thumbs, trade, secondReferrerAward);
            } catch (Exception e) {
                log.error("", e);
                stat.setRollbackOnly();
                return MessageResult.error(e.getMessage());
            }
        });

        // 成功
        if (result.getCode() == 0) {
            // TODO: 保存交易委托详情
        }

        log.info("trade {}:{} {} ${} result = {} msg = {}", trade.getBuyOrderId(), trade.getSellOrderId(),
                trade.getAmount(), trade.getPrice(), result.getCode() == 0, result.getMessage());

        if (result.getCode() == 0) {
            successTrades = Collections.singletonList(trade);
        }

        ExchangeOrder buyOrder = exchangeOrderService.findOne(trade.getBuyOrderId());
        ExchangeOrder sellOrder = exchangeOrderService.findOne(trade.getSellOrderId());
        try {
            // 推送订单成交订阅
            messagingTemplate.convertAndSend(
                    "/topic/market/order-trade/" + symbol + "/" + buyOrder.getMemberId(), buyOrder);
            messagingTemplate.convertAndSend(
                "/topic/market/order-trade/" + symbol + "/" + sellOrder.getMemberId(), sellOrder);

            //todo 就是告诉前端，发个通知而已，客户端没有处理数据，客户端只是刷新了列表
            nettyHandler.handleOrder(NettyCommand.PUSH_EXCHANGE_ORDER_TRADE, buyOrder);
            nettyHandler.handleOrder(NettyCommand.PUSH_EXCHANGE_ORDER_TRADE, sellOrder);

//            nettyPushDataUtil.pushMarketTrade(symbol, Lists.newArrayList(buyOrder));
//            nettyPushDataUtil.pushMarketTrade(symbol,Lists.newArrayList(sellOrder));
        } catch (Exception e) {
            log.error("", e);
        }

        // 处理K线行情
        if (coinProcessor != null) {
            try {
                coinProcessor.process(successTrades);
            } catch (Exception e) {
                log.error("", e);
            }
        }
        pushJob.addTrades(symbol, successTrades);
    }

    @SneakyThrows
    private void handleTrades(List<ExchangeTrade> trades) {
        for (ExchangeTrade trade : trades) {
            handleTrade(trade);
        }
    }

    @KafkaListener(topics = "exchange-order-completed", containerFactory = "kafkaListenerContainerFactory")
    public void handleOrderCompleted(List<ConsumerRecord<String, String>> records) {
        try {
            for (int i = 0; i < records.size(); i++) {
                ConsumerRecord<String, String> record = records.get(i);
                List<ExchangeOrder> orders = JSON.parseArray(record.value(), ExchangeOrder.class);
                for (ExchangeOrder order : orders) {
                    handleComplete(order);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @KafkaListener(topics = "exchange-trade-result")
    public void handleResult(List<ConsumerRecord<String, String>> records) {
        for (int i = 0; i < records.size(); i++) {
            ConsumerRecord<String, String> record = records.get(i);
            String result = record.value();

            try {
                // 处理交易
                if (result.contains("\"buyOrderId\"")) {
                    ExchangeTrade trade = JSON.parseObject(result, ExchangeTrade.class);
                    handleTrade(trade);
                    // 处理撤单或者订单完成
                } else {
                    ExchangeOrder order = JSON.parseObject(result, ExchangeOrder.class);
                    switch (order.getStatus()) {
                        case CANCELED: {
                            handleCancel(order);
                            break;
                        }
                        case COMPLETED: {
                            handleComplete(order);
                            break;
                        }
                        default: {
                            log.error("unexpected order status {}", order.getStatus());
                        }
                    }
                }
            } catch (Exception ex) {
                log.error("", ex);
            }
        }
    }

    /**
     * 消费交易盘口信息
     *
     * @param records
     */
    @KafkaListener(topics = "exchange-trade-plate", containerFactory = "kafkaListenerContainerFactory")
    public void handleTradePlate(List<ConsumerRecord<String, String>> records) {
        try {
            for (int i = 0; i < records.size(); i++) {
                ConsumerRecord<String, String> record = records.get(i);
                TradePlate plate = JSON.parseObject(record.value(), TradePlate.class);
                String symbol = plate.getSymbol();
                pushJob.addPlates(symbol, plate);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @KafkaListener(topics = "exchange-order-cancel-success", containerFactory = "kafkaListenerContainerFactory")
    public void handleOrderCanceled(List<ConsumerRecord<String, String>> records) {
        try {
            for (int i = 0; i < records.size(); i++) {
                ConsumerRecord<String, String> record = records.get(i);
                ExchangeOrder order = JSON.parseObject(record.value(), ExchangeOrder.class);
                handleCancel(order);
            }
        } catch (Exception e) {
            log.error("", e);
        }
    }

    public class HandleTradeThread implements Runnable {
        private ConsumerRecord<String, String> record;

        private HandleTradeThread(ConsumerRecord<String, String> record) {
            this.record = record;
        }

        @Override
        public void run() {
            try {
                List<ExchangeTrade> trades = JSON.parseArray(record.value(), ExchangeTrade.class);
                handleTrades(trades);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
